package com.llx.datastream

import java.sql.{Connection, DriverManager, PreparedStatement}
import java.util

import com.llx.datastream.自定义Source_mysql.Student
import org.apache.calcite.avatica.proto.Requests
import org.apache.flink.api.common.functions.RuntimeContext
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction
import org.apache.flink.streaming.api.scala.{StreamExecutionEnvironment, _}
import org.apache.flink.streaming.util.serialization.SimpleStringSchema
object 自定义sink_mysql {
  def main(args: Array[String]): Unit = {

    //1.创建流执行环境
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    //2.准备数据
    val dataStream:DataStream[Student] = env.fromElements(
      Student(5, "dahua", "beijing biejing", "female"),
      Student(6, "daming", "tainjing tianjin", "male "),
      Student(7, "daqiang ", "shanghai shanghai", "female")
    )

    //3.将数据写入到自定义的sink中（这里是mysql）
    dataStream.addSink(new StudentSinkToMysql)

    //4.触发流执行
    env.execute()
  }

  /**
    * mysql sink
    */
  class StudentSinkToMysql extends RichSinkFunction[Student] {
    private var connection: Connection = null
    private var ps: PreparedStatement = null

    /**
      * 一、open()方法中建立连接，这样不用每次invoke的时候都要建立连接和释放连接。
      */
    override def open(parameters: Configuration): Unit = {
      super.open(parameters)
      val driver = "com.mysql.jdbc.Driver"
      val url = "jdbc:mysql://qingcheng11:3306/flinktest"
      val username = "root"
      val password = "qingcheng"
      //1.加载驱动
      Class.forName(driver)
      //2.创建连接
      connection = DriverManager.getConnection(url, username, password)
      val sql = "insert into Student(stuid,stuname,stuaddr,stusex)values(?,?,?,?);"
      //3.获得执行语句
      ps = connection.prepareStatement(sql)
    }

    /**
      * 二、每个元素的插入都要调用一次invoke()方法，这里主要进行插入操作
      */
    override def invoke(stu: Student): Unit = {
      try {
        //4.组装数据，执行插入操作
        ps.setInt(1, stu.stuid)
        ps.setString(2, stu.stuname)
        ps.setString(3, stu.stuaddr)
        ps.setString(4, stu.stusex)
        ps.executeUpdate()
      } catch {
        case e: Exception => println(e.getMessage)
      }
    }

    /**
      * 三、 程序执行完毕就可以进行，关闭连接和释放资源的动作了
      */
    override def close(): Unit = {
      super.close()
      //5.关闭连接和释放资源
      if (connection != null) {
        connection.close()
      }
      if (ps != null) {
        ps.close()
      }
    }
  }


  /**
    * Elasticsearch连接器
    */
  /*def ex(): Unit ={
    val config = new util.HashMap[String, String]
    config.put("bulk.flush.max.actions", "1")
    config.put("cluster.name", "my-cluster-name")
    env.readTextFile("")
      .addSink(new ElasticsearchSink(config, new IndexRequestBuilder[String] {
        override def createIndexRequest(element: String, ctx: RuntimeContext): IndexRequest = {
          val json = new util.HashMap[String, AnyRef]
          json.put("data", element)
          println("SENDING: " + element)
          Requests.indexRequest.index("my-index").`type`("my-type").source(json)
        }
      }))
  }*/



  /**
    * RabbitMQ连接器
    */
/*  def ddd(): Unit ={
    val stream = env
      .addSource(new RMQSource[String](
        connectionConfig,            // config for the RabbitMQ connection
        "queueName",                 // name of the RabbitMQ queue to consume
        true,                        // use correlation ids; can be false if only at-least-once is required
        new SimpleStringSchema))
  }*/

}
